package com.atguigu.gmall.publisher.service.impl;

import com.atguigu.gmall.publisher.bean.DetailItemInfo;
import com.atguigu.gmall.publisher.bean.NameValue;
import com.atguigu.gmall.publisher.service.OrderRealtimeService;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedSum;
import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Service
public class OrderRealtimeServiceImpl implements OrderRealtimeService {

    @Autowired
    RestHighLevelClient restHighLevelClient;

    public static final String ORDER_INDEX_PREFIX="gmall0819_order_wide_";

    @Override
    public List<NameValue> statsByItem(String itemName, String date, String type) {


        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(ORDER_INDEX_PREFIX+date);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        // 查询条件

        // 匹配
        searchSourceBuilder.query(new MatchQueryBuilder("sku_name",itemName).operator(Operator.AND));

        //size
        searchSourceBuilder.size(0);
        //聚合
        //  分组
        TermsAggregationBuilder termAggBuilder =getTermAggBuilderByType(  type);
        // 聚合
        SumAggregationBuilder sumAggBuilder = AggregationBuilders.sum("sum_amount").field("split_total_amount");
        // 分组+聚合
        termAggBuilder.subAggregation(sumAggBuilder);
        searchSourceBuilder.aggregation(termAggBuilder);

        searchRequest.source(searchSourceBuilder);

        try {
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            //提取分组结果
            ParsedTerms  termsAgg = (ParsedTerms)searchResponse.getAggregations().get(getTermNameByType(type));
            List<? extends Terms.Bucket> buckets = termsAgg.getBuckets();

            List<NameValue> nameValueList=new ArrayList<>();
            //迭代分组结果
            for (Terms.Bucket bucket : buckets) {
                String key = String.valueOf( bucket.getKey());
                //提取每个组的聚合值
                ParsedSum sumAgg = (ParsedSum)bucket.getAggregations().get("sum_amount");
                double sumValue = sumAgg.getValue();
                nameValueList.add(  new NameValue(key, BigDecimal.valueOf(sumValue)));
            }
            return trimResultByType(type,nameValueList);

        } catch (IOException e) {
            e.printStackTrace();
            throw  new RuntimeException("访问es 异常");
        }catch (ElasticsearchStatusException ese){
            if( ese.status()== RestStatus.NOT_FOUND  ){
                return  new ArrayList<>();
            }
        }


        return null;
    }

    @Override
    public DetailItemInfo detailByItem(String itemName, String date, Integer pageNo, Integer pageSize) {
       //查询条件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        // 匹配
        searchSourceBuilder.query(new MatchQueryBuilder("sku_name",itemName).operator(Operator.AND));

        //分页
        searchSourceBuilder.size(pageSize);
        //行号= 页码-1 ）*每页行数
        searchSourceBuilder.from( (pageNo-1)*pageSize );
        //抓取字段
        searchSourceBuilder.fetchSource( new String[]{"create_time", "sku_name", "order_price", "province_name", "sku_num", "total_amount", "user_age", "user_gender"}, null);
        //高亮
        searchSourceBuilder.highlighter(new HighlightBuilder().field("sku_name"));

        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(ORDER_INDEX_PREFIX+date);
        searchRequest.source(searchSourceBuilder);

        try {
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
           //提取结果
            //明细结果
            SearchHit[] searchHits = searchResponse.getHits().getHits();
            List<Map> mapList=new ArrayList<>(searchHits.length);
            for (int i = 0; i < searchHits.length; i++) {
                SearchHit searchHit = searchHits[i];
                Map<String, Object> sourceAsMap = searchHit.getSourceAsMap();

                //高亮字段提取
                HighlightField highlightField = searchHit.getHighlightFields().get("sku_name");
                Text text = highlightField.getFragments()[0];
                String skuNameHL = text.string();
                sourceAsMap.put("sku_name",skuNameHL);

                mapList.add(sourceAsMap);
            }
            //总数
            long total = searchResponse.getHits().getTotalHits().value;
            return new DetailItemInfo(total,mapList);


        } catch (IOException e) {
            e.printStackTrace();
            throw  new RuntimeException("访问es 异常");
        }catch (ElasticsearchStatusException ese){
            if( ese.status()== RestStatus.NOT_FOUND  ){
                return  new DetailItemInfo();
            }
        }
            return null;
    }

    // private List<NameValue>  statsByItem()

    private  String  getTermNameByType(String type){
        if("gender".equals(type)){
            return "groupby_gender";
        }else if("age".equals(type)){
               return "groupby_age";
        }else{
            throw new RuntimeException("不支持该类型");
        }
    }

    // 跟据类型整理返回的结果
    private  List<NameValue>  trimResultByType(String type,List<NameValue> nameValueList){
            if("gender".equals(type)){
                return nameValueList;
            }else if("age".equals(type)){
                //TODO 把年龄聚合 转为年龄段聚合
                return nameValueList;
            }
            return  null;
    }


        private  TermsAggregationBuilder  getTermAggBuilderByType(String type){
           if("gender".equals(type)){
               TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders.terms("groupby_gender").field("user_gender").size(2);
               return termsAggregationBuilder;
           }else if("age".equals(type)){
               TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders.terms("groupby_age").field("user_age").size(100);
               return termsAggregationBuilder;
           }else{
               throw new RuntimeException("不支持该类型");
           }
        }

}
